import com.hankcs.hanlp.HanLP
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SouGouDemo {
  def main(args: Array[String]): Unit = {
    //TODO 0.准备环境
    val conf: SparkConf = new SparkConf().setAppName("spark").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //1,加载数据
    val lines: RDD[String] = sc.textFile("data/SogouQ.sample")
    //2,处理数据
    val sogouRecordRdd: RDD[SogouRecord] = lines.map(line => {
      val arr: Array[String] = line.split("\\s+")
      SogouRecord(
        arr(0),
        arr(1),
        arr(2),
        arr(3).toInt,
        arr(4).toInt,
        arr(5)
      )
    })
//    切割数据
    val words: RDD[String] = sogouRecordRdd.flatMap(record => {
      val wordsStr: String = record.queryWords.replaceAll("\\[|\\]", "") //360安全卫士
      import scala.collection.JavaConverters._ //将Java集合转为scala集合
      HanLP.segment(wordsStr).asScala.map(_.word) //ArrayBuffer(360, 安全卫士)
    })

//    println(words.collect().take(20).toBuffer)
    //热门搜索词统计
    val result: Array[(String, Int)] = words
      .filter(word => !word.equals(".") && !word.equals("+")&& !word.equals("的"))
      .map((_, 1)).reduceByKey(_ + _).sortBy(_._2, false).take(3)
    println(result.toBuffer)




  }

  case class SogouRecord(
                          queryTime: String,
                          userId: String,
                          queryWords: String,
                          resultRank: Int,
                          clickRank: Int,
                          clickUrl: String
                        )

}
